﻿using Confluent.Kafka;
using Grow.Binary;
using Grow.EventBus.Cap.DomainEventSend;
using Grow.EventBus.Cap.PublisherDomainEvents;
using Grow.Json;
using Grow.Module.Service;
using Grow.Utils.Exceptions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Grow.EventBus.Cap.Kafka
{
    public class DomainEventSend : IDomainEventSend
    {
        private readonly IServiceContext _serviceContext;
        private readonly IJsonContext _jsonContext;
        private readonly IKafkaConnection _kafkaConnection;
        private readonly IBinaryContext _binaryContext;
        public DomainEventSend(IServiceContext serviceContext)
        {
            _serviceContext = serviceContext;
            _jsonContext = _serviceContext.GetRequiredService<IJsonContext>();
            _kafkaConnection = _serviceContext.GetRequiredService<IKafkaConnection>();
            _binaryContext = _serviceContext.GetRequiredService<IBinaryContext>();
        }
        public async Task SendAsync(PublisherDomainEventDto publisherDomainEventDto, CancellationToken cancellationToken = default)
        {
            var producer =  _kafkaConnection.GetProducer(publisherDomainEventDto.EventName);


            var value =await _binaryContext.SerializeAsync(publisherDomainEventDto, cancellationToken);
            if (value is null)
            {
                throw new GrowCheckException($"[EventBus-Kafka] send msg fail, produce eventDataDto is null, {publisherDomainEventDto.Id}");
            }
            var result = await producer.ProduceAsync(publisherDomainEventDto.EventName, new Message<string, byte[]>
            {
                Key = publisherDomainEventDto.Id.ToString(),
                Value = value
            }, cancellationToken);

            if (!(result.Status is PersistenceStatus.Persisted or PersistenceStatus.PossiblyPersisted))
            {
                throw new GrowCheckException($"[EventBus-Kafka] send msg fail, produce status  {result.Status}  {publisherDomainEventDto.Id}");
            }
        }
    }
}
